package org.raymond.iworks.study.mq.rabbitmq.classic.topics;

import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;
import org.raymond.iworks.study.mq.rabbitmq.classic.hello.Send;

import java.nio.charset.StandardCharsets;

/**
 * @author raymond
 * @version V1.0
 * @Description: TODO
 */
@Slf4j
public class ReceiveLogsTopic {

    //* (star) 只能代替一个词.
    //# (hash) 可以替代0个或多个单词
    // args1. "#"
    // args2. "kern.*"
    // args3. "*.critical"
    // args4. "kern.*" "*.critical"
    public static void main(String[] args) throws Exception {
        String EXCHANGE_NAME = EmitLogTopic.EXCHANGE_NAME;
        Connection conn = Send.connection();
        Channel channel = conn.createChannel();

        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
        String queueName = channel.queueDeclare().getQueue();

        if(args.length<1){
            log.error("Usage: ReceiveLogsTopic [binding_key]...");
            System.exit(1);
        }
        for(String bindingKey : args) {
            channel.queueBind(queueName, EXCHANGE_NAME, bindingKey);
        }
        log.info("[*] Waiting for message");

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
            log.info(" [x] Received {}:{}", delivery.getEnvelope().getRoutingKey(), message);
            try{
                doWork(message);
            }catch (InterruptedException e){
                log.error("{}: {} will intterrupt.", e.getMessage(), Thread.currentThread().getName());
                Thread.currentThread().interrupt();
            }finally {
                log.info(" [x] Done");
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        };

        boolean autoAck = false;
        channel.basicConsume(queueName, autoAck, deliverCallback, consumerTag->{});
    }

    private static void doWork(String task) throws InterruptedException {
        if(task.startsWith("interrupted")){
            Thread.currentThread().interrupt();
        }
        for (char ch: task.toCharArray()){
            if(ch=='.') Thread.sleep(1000);
        }
    }
}
